
package lease

import (
	"container/heap"
	"context"
	"encoding/binary"
	"errors"
	"math"
	"sort"
	"sync"
	"time"

	"github.com/coreos/go-semver/semver"
	pb "go.etcd.io/etcd/api/v3/etcdserverpb"
	"go.etcd.io/etcd/server/v3/lease/leasepb"
	"go.etcd.io/etcd/server/v3/storage/backend"
	"go.etcd.io/etcd/server/v3/storage/schema"
	"go.uber.org/zap"
)

// NoLease是一种特殊租赁，表示没有租赁。
const NoLease = LeaseID(0)

// MaxLeaseTTL是最大租约TTL值
const MaxLeaseTTL = 9000000000

var v3_6 = semver.Version{Major: 3, Minor: 6}

var (
	forever = time.Time{}

	// 每秒要撤销的最大租约数；可配置用于测试
	leaseRevokeRate = 1000

	// 每秒记录到一致性日志的最大租赁检查点数；可配置用于测试
	leaseCheckpointRate = 1000

	// 租约检查点
	defaultLeaseCheckpointInterval = 5 * time.Minute

	// 批处理到单个一致日志项中的租约检查点的最大数量
	maxLeaseCheckpointBatchSize = 1000

	// 检查是否撤销过期租约的默认间隔
	defaultExpiredleaseRetryInterval = 3 * time.Second

	ErrNotPrimary       = errors.New("not a primary lessor")
	ErrLeaseNotFound    = errors.New("lease not found")
	ErrLeaseExists      = errors.New("lease already exists")
	ErrLeaseTTLTooLarge = errors.New("too large lease TTL")
)

// TxnDelete是只允许删除的TxnWrite。此处定义
// 以避免与mvcc的循环依赖关系。
type TxnDelete interface {
	DeleteRange(key, end []byte) (n, rev int64)
	End()
}

// RangeDeleter是一个TxnDelete构造函数。
type RangeDeleter func() TxnDelete

// 检查点程序允许对协商一致日志中剩余的TTL进行检查。在这里定义
// 避免与mvcc的循环依赖。
type Checkpointer func(ctx context.Context, lc *pb.LeaseCheckpointRequest)

type LeaseID int64

// 在 etcd v2 的存储时看到，非永久节点中都记录了节点的存活时间
// (CExpireTime ），并且通过 ttlKeyHeap 将节点按照过期时间进行了排序。在介绍etcd 服务
// 端时会看到，服务端会定时发送 SYNC 请求， 该请求主要用于触发 v2 存储调用前面分析的
// DeleteExpiredKeys （）方法清理过期的节点。
// 在etcd v3 存储中并没有提供类似的过期时间的设置，而是通过 Lease （租约〉实现键值对过
// 期的效果。在 Lessor 接口中定义了授权、撤销、修改 Lease 的相关方法，
type Lessor interface {
	SetRangeDeleter(rd RangeDeleter)
	SetCheckpointer(cp Checkpointer)
	// 创建 Lease 实例 ， 该 Lease 会在指定的时间（ ttl ）之后过期
	Grant(id LeaseID, ttl int64) (*Lease, error)
	// 撤销指定的 Lease ， 该 Lease 实例相关的 LeaseItem 也会被删除
	Revoke(id LeaseID) error
	Checkpoint(id LeaseID, remainingTTL int64) error
	// 将指定 Lease 与指定 LeaseItem 绑定，在 LeaseItem 中封装了键位对的 Key 位
	Attach(id LeaseID, items []LeaseItem) error
	// 根据 LeaseItem 查询对应 Lease 实例 的 id
	GetLease(item LeaseItem) LeaseID
	// 取消指定 Lease 与 指定 LeaseItem 之间的绑定关 系
	Detach(id LeaseID, items []LeaseItem) error
	// 如果当前节点成为 Leader 节 点，则其使用的 Lessor 实例将通过该方法晋升，成为主 Lessor
	// 主 Lessor 将控制着整个集群中所有 Lease 实例
	Promote(extend time.Duration)
	// 如果当前节点从 Leader 状态转换为其他状态，则会通过该方法将其使用的 Lessor 实例进行降级
	Demote()
	// 续约指定的 lease 实例
	Renew(id LeaseID) (int64, error)
	// 查找指定 id 对应的 Lease 实例
	Lookup(id LeaseID) *Lease
	Leases() []*Lease
	// 如果出现 lease 实例过期 ， 则会被写入圭1J ExpiredLeasesC 通道中
	ExpiredLeasesC() <-chan []*Lease
	// 从底层的 v3 存储中恢复 Lease 实例
	Recover(b backend.Backend, rd RangeDeleter)
	Stop()
}

// 出租方实现出租方接口。
// TODO:使用时钟系统实现可测试性。
type lessor struct {
	mu sync.RWMutex
	demotec chan struct{}
	// 记录 id 到 Lease 实例之间的映射
	leaseMap             map[LeaseID]*Lease
	leaseExpiredNotifier *LeaseExpiredNotifier
	leaseCheckpointHeap  LeaseQueue
	// 记录 Lease Ite m 到 Lease 实例 i d 的映射
	itemMap              map[LeaseItem]LeaseID
	// Range Deleter 接口主要用于从底层的存储中删除过期的 lease 实例。
	rd RangeDeleter
	cp Checkpointer
	// 底层持久化 Lease 的存储
	b backend.Backend
	// Lease 实例过期时间的最小值。
	minLeaseTTL int64
	// 过期的 Lease 实例会被写入该通道中，并等待其他 goroutine 进行处理。
	expiredC chan []*Lease
	stopC chan struct{}
	// 用于判断当前 lessor 实例是否为主 lessor( primary lessor ） 。
	// 如果当前 lessor 是主 Lessor 实例，则会开启该通道，当节点切换成非 Leader 状态时，
	// 会关闭该通道。
	doneC chan struct{}
	lg *zap.Logger
	checkpointInterval time.Duration
	expiredLeaseRetryInterval time.Duration
	checkpointPersist bool
	cluster cluster
}

type cluster interface {
	// 版本是集群范围内的最小主版本。次要版本。
	Version() *semver.Version
}

type LessorConfig struct {
	MinLeaseTTL                int64
	CheckpointInterval         time.Duration
	ExpiredLeasesRetryInterval time.Duration
	CheckpointPersist          bool
}

func NewLessor(lg *zap.Logger, b backend.Backend, cluster cluster, cfg LessorConfig) Lessor {
	return newLessor(lg, b, cluster, cfg)
}

// 在 newLessor（）函数中会创建 lessor 实例，并调用 initAndRecover（）方法完成初始化操作，同
// 时还会启动一个后台 goroutine 查找当前 lessor 中是否存在过期的 lease 实例 。
func newLessor(lg *zap.Logger, b backend.Backend, cluster cluster, cfg LessorConfig) *lessor {
	checkpointInterval := cfg.CheckpointInterval
	expiredLeaseRetryInterval := cfg.ExpiredLeasesRetryInterval
	if checkpointInterval == 0 {
		checkpointInterval = defaultLeaseCheckpointInterval
	}
	if expiredLeaseRetryInterval == 0 {
		expiredLeaseRetryInterval = defaultExpiredleaseRetryInterval
	}
	l := &lessor{
		leaseMap:                  make(map[LeaseID]*Lease),
		itemMap:                   make(map[LeaseItem]LeaseID),
		leaseExpiredNotifier:      newLeaseExpiredNotifier(),
		leaseCheckpointHeap:       make(LeaseQueue, 0),
		b:                         b,
		minLeaseTTL:               cfg.MinLeaseTTL,
		checkpointInterval:        checkpointInterval,
		expiredLeaseRetryInterval: expiredLeaseRetryInterval,
		checkpointPersist:         cfg.CheckpointPersist,
		// expiredC是一个小的缓冲chan，以避免不必要的阻塞。
		expiredC: make(chan []*Lease, 16),
		stopC:    make(chan struct{}),
		doneC:    make(chan struct{}),
		lg:       lg,
		cluster:  cluster,
	}
	l.initAndRecover()
	// 在 newLessor（）函 数中启动的后台 goroutine 执行的是 lessor.runLoop（） 方法，其中会根据当前
	// lessor 实例是否为主 lessor ， 决定是否检测过期 Lease 实例，
	go l.runLoop()

	return l
}

// isPrimary表示此出租人是否为主要出租人。主
// 出租人管理租赁到期和续租。
//
// 在etcd中，筏长是主要的。因此，可能同时有两位主要领导人（raft允许同时担任领导人，但任期不同）
// 最多一次领导人选举超时。
// 旧的主要领导人不能影响正确性，因为其提案的期限较短，不会承诺。
//
// TODO:筏式跟随者不提交租赁管理提案。
// 可能存在一个非常小的窗口（通常在秒内，这取决于go调度），
// 在筏长降级和出租人降级之间，筏跟随是主要的。
// 通常这应该不是问题。租赁不应该对时间如此敏感。
func (le *lessor) isPrimary() bool {
	return le.demotec != nil
}

func (le *lessor) SetRangeDeleter(rd RangeDeleter) {
	le.mu.Lock()
	defer le.mu.Unlock()

	le.rd = rd
}

func (le *lessor) SetCheckpointer(cp Checkpointer) {
	le.mu.Lock()
	defer le.mu.Unlock()

	le.cp = cp
}

// lessor.Grant（）方法会根据指定的 id 和过期时长新建 Lease 实例 ，然后将其保存到 leaseMap
// 中并进行持久化，
func (le *lessor) Grant(id LeaseID, ttl int64) (*Lease, error) {
	if id == NoLease {
		return nil, ErrLeaseNotFound
	}

	if ttl > MaxLeaseTTL {
		return nil, ErrLeaseTTLTooLarge
	}

	// 新建 Lease 实例
	l := &Lease{
		ID:      id,
		ttl:     ttl,
		itemSet: make(map[LeaseItem]struct{}),
		revokec: make(chan struct{}),
	}

	le.mu.Lock()
	defer le.mu.Unlock()
	// 检测指定的id 在当前 leaseMap 中是否存在对应 Lease 实例
	if _, ok := le.leaseMap[id]; ok {
		return nil, ErrLeaseExists
	}
	// 修正过期时长
	if l.ttl < le.minLeaseTTL {
		l.ttl = le.minLeaseTTL
	}

	if le.isPrimary() {
		l.refresh(0)	// 更新 Lease 实例的过期时间戳
	} else {
		l.forever()	// 如果当前节 点不是 Leader ，则 将 Lease 设置为永不过期
	}

	le.leaseMap[id] = l	// 在 leaseMap 中记录该 Lease 实例
	l.persistTo(le.b)	// 持久化 Lease 信息

	leaseTotalTTLs.Observe(float64(l.ttl))
	leaseGranted.Inc()

	if le.isPrimary() {
		item := &LeaseWithTime{id: l.ID, time: l.expiry}
		le.leaseExpiredNotifier.RegisterOrUpdate(item)
		le.scheduleCheckpointIfNeeded(l)
	}

	return l, nil
}

// lessor.Revoke（）方法负责撤销指定 id 对应的 Lease 实例， 其中会关闭 Lease 实例对应的
// revokec 通道，并将 Lease 实例从 lessor.leaseMap 和存储中删除 。需要注意的是，除了删除 Lease
// 实例本身，还需要删除与该 Lease 实例关联的键值对。
func (le *lessor) Revoke(id LeaseID) error {
	le.mu.Lock()

	l := le.leaseMap[id]	// 从 leaseMap 中 查找指定的 Lease 实例
	if l == nil {
		le.mu.Unlock()
		return ErrLeaseNotFound
	}
	defer close(l.revokec)	// 在方法结束时， 关闭该 Lease 实例中的 revokec 远远
	// 在执行外部工作之前解锁
	le.mu.Unlock()

	if le.rd == nil {
		return nil
	}

	txn := le.rd()

	keys := l.Keys()	// 获取与该 Lease 实例绑定的 Key 值
	sort.StringSlice(keys).Sort()
	for _, key := range keys {
		txn.DeleteRange([]byte(key), nil)	// 删除与该 Lease 实例绑定的全部键值对
	}

	le.mu.Lock()
	defer le.mu.Unlock()
	delete(le.leaseMap, l.ID)	// 将该 Lease 实例从 leaseMap 中删除
	// 将该 Lease 实例的信息从底层的 v3 存储中删除
	schema.UnsafeDeleteLease(le.b.BatchTx(), &leasepb.Lease{ID: int64(l.ID)})

	txn.End()	// 提交事务

	leaseRevoked.Inc()
	return nil
}

func (le *lessor) Checkpoint(id LeaseID, remainingTTL int64) error {
	le.mu.Lock()
	defer le.mu.Unlock()

	if l, ok := le.leaseMap[id]; ok {
		// 检查点时，我们只更新剩余的TTL，升级负责将此应用于租约到期
		l.remainingTTL = remainingTTL
		if le.shouldPersistCheckpoints() {
			l.persistTo(le.b)
		}
		if le.isPrimary() {
			// 根据需要安排下一个检查点
			le.scheduleCheckpointIfNeeded(l)
		}
	}
	return nil
}

func (le *lessor) shouldPersistCheckpoints() bool {
	cv := le.cluster.Version()
	return le.checkpointPersist || (cv != nil && greaterOrEqual(*cv, v3_6))
}

func greaterOrEqual(first, second semver.Version) bool {
	return !first.LessThan(second)
}

// lessor.Renew （）方法负责续租一个己经存在的 Lease 实例，当然，只有当前节点是 Leader 时
// 才能完成该续租操作。
func (le *lessor) Renew(id LeaseID) (int64, error) {
	le.mu.RLock()
	// 检测当前 l essor 是否为主 lessor 实例
	if !le.isPrimary() {
		le.mu.RUnlock()
		return -1, ErrNotPrimary
	}

	demotec := le.demotec

	l := le.leaseMap[id]	// 在 leaseMap 中查找指定的 Lease 实例
	// 如果指定 Lease 不存在，则返回错误信息
	if l == nil {
		le.mu.RUnlock()
		return -1, ErrLeaseNotFound
	}
	// 如果设置为
	clearRemainingTTL := le.cp != nil && l.remainingTTL > 0

	le.mu.RUnlock()
	if l.expired() {	// 检测 Lease 实例是否过期
		// 如果该 Lease 已经过期，如l 返回错误信息（
		select {
		case <-l.revokec:
			return -1, ErrLeaseNotFound
		case <-demotec:
			return -1, ErrNotPrimary
		case <-le.stopC:
			return -1, ErrNotPrimary
		}
	}

	// 如果设置了剩余TTL，则在续订时清除剩余TTL
	// 通过仅在已设置剩余TTL时应用RAFT条目，我们将每次租约写入的RAFT条目数
	// 限制为每个检查点间隔最多2个。
	if clearRemainingTTL {
		le.cp(context.Background(), &pb.LeaseCheckpointRequest{Checkpoints: []*pb.LeaseCheckpoint{{ID: int64(l.ID), Remaining_TTL: 0}}})
	}

	le.mu.Lock()
	l.refresh(0)	// 更新过期时间
	item := &LeaseWithTime{id: l.ID, time: l.expiry}
	le.leaseExpiredNotifier.RegisterOrUpdate(item)
	le.mu.Unlock()

	leaseRenewed.Inc()
	return l.ttl, nil
}

func (le *lessor) Lookup(id LeaseID) *Lease {
	le.mu.RLock()
	defer le.mu.RUnlock()
	return le.leaseMap[id]
}

func (le *lessor) unsafeLeases() []*Lease {
	leases := make([]*Lease, 0, len(le.leaseMap))
	for _, l := range le.leaseMap {
		leases = append(leases, l)
	}
	return leases
}

func (le *lessor) Leases() []*Lease {
	le.mu.RLock()
	ls := le.unsafeLeases()
	le.mu.RUnlock()
	sort.Sort(leasesByExpiry(ls))
	return ls
}

func (le *lessor) Promote(extend time.Duration) {
	le.mu.Lock()
	defer le.mu.Unlock()

	le.demotec = make(chan struct{})

	// 刷新所有租约的到期日。
	for _, l := range le.leaseMap {
		l.refresh(extend)
		item := &LeaseWithTime{id: l.ID, time: l.expiry}
		le.leaseExpiredNotifier.RegisterOrUpdate(item)
		le.scheduleCheckpointIfNeeded(l)
	}

	if len(le.leaseMap) < leaseRevokeRate {
		// 没有租赁堆积的可能性
		return
	}

	// 在重叠的情况下调整到期日
	leases := le.unsafeLeases()
	sort.Sort(leasesByExpiry(leases))

	baseWindow := leases[0].Remaining()
	nextWindow := baseWindow + time.Second
	expires := 0
	// 到期日少于总撤销率，因此堆积租赁
	// 不消耗整个撤销限制
	targetExpiresPerSecond := (3 * leaseRevokeRate) / 4
	for _, l := range leases {
		remaining := l.Remaining()
		if remaining > nextWindow {
			baseWindow = remaining
			nextWindow = baseWindow + time.Second
			expires = 1
			continue
		}
		expires++
		if expires <= targetExpiresPerSecond {
			continue
		}
		rateDelay := float64(time.Second) * (float64(expires) / float64(targetExpiresPerSecond))
		// 如果租赁延长n秒，
		// 基本窗口前的n秒应仅延长一秒。
		rateDelay -= float64(remaining - baseWindow)
		delay := time.Duration(rateDelay)
		nextWindow = baseWindow + delay
		l.refresh(delay + extend)
		item := &LeaseWithTime{id: l.ID, time: l.expiry}
		le.leaseExpiredNotifier.RegisterOrUpdate(item)
		le.scheduleCheckpointIfNeeded(l)
	}
}

type leasesByExpiry []*Lease

func (le leasesByExpiry) Len() int           { return len(le) }
func (le leasesByExpiry) Less(i, j int) bool { return le[i].Remaining() < le[j].Remaining() }
func (le leasesByExpiry) Swap(i, j int)      { le[i], le[j] = le[j], le[i] }

func (le *lessor) Demote() {
	le.mu.Lock()
	defer le.mu.Unlock()

	// 将所有租约的到期日设置为永久
	for _, l := range le.leaseMap {
		l.forever()
	}

	le.clearScheduledLeasesCheckpoints()
	le.clearLeaseExpiredNotifier()

	if le.demotec != nil {
		close(le.demotec)
		le.demotec = nil
	}
}

// lessor.Attach（） 方法负责将指定的 Lease 实例和键值对绑定，
func (le *lessor) Attach(id LeaseID, items []LeaseItem) error {
	le.mu.Lock()
	defer le.mu.Unlock()

	l := le.leaseMap[id]	// 从 leaseMap 中查找指定的 Lease 实例
	if l == nil {
		return ErrLeaseNotFound
	}

	l.mu.Lock()
	for _, it := range items {
		l.itemSet[it] = struct{}{}	// 在 itemSet 这个 map 中记录该 LeaseItem 实例
		le.itemMap[it] = id	// 在 Lease 实例的工temMap 字段中记录该 LeaseItem 实例，从而实现绑定
	}
	l.mu.Unlock()
	return nil
}

func (le *lessor) GetLease(item LeaseItem) LeaseID {
	le.mu.RLock()
	id := le.itemMap[item]
	le.mu.RUnlock()
	return id
}

// Detach（） 方法负责解绑指定的 Lease 实例和键值对解绑
func (le *lessor) Detach(id LeaseID, items []LeaseItem) error {
	le.mu.Lock()
	defer le.mu.Unlock()

	l := le.leaseMap[id]
	if l == nil {
		return ErrLeaseNotFound
	}

	l.mu.Lock()
	for _, it := range items {
		delete(l.itemSet, it)
		delete(le.itemMap, it)
	}
	l.mu.Unlock()
	return nil
}

func (le *lessor) Recover(b backend.Backend, rd RangeDeleter) {
	le.mu.Lock()
	defer le.mu.Unlock()

	le.b = b
	le.rd = rd
	le.leaseMap = make(map[LeaseID]*Lease)
	le.itemMap = make(map[LeaseItem]LeaseID)
	le.initAndRecover()
}

func (le *lessor) ExpiredLeasesC() <-chan []*Lease {
	return le.expiredC
}

func (le *lessor) Stop() {
	close(le.stopC)
	<-le.doneC
}

// 在 newLessor（）函数中启动 的后台 goroutine 执行的是 lessor.runLoop（）方法，其中会根据当前
//lessor 实例是否为主 lessor ， 决定是否检测过期 Lease 实例，
func (le *lessor) runLoop() {
	defer close(le.doneC)

	for {
		le.revokeExpiredLeases()
		le.checkpointScheduledLeases()

		select {
		case <-time.After(500 * time.Millisecond):
		case <-le.stopC:
			return
		}
	}
}

// revokeExpiredLeases查找所有过期的租约，并将其发送到过期频道，以便
// 被撤销。
func (le *lessor) revokeExpiredLeases() {
	var ls []*Lease

	// 利率限制
	revokeLimit := leaseRevokeRate / 2

	le.mu.RLock()
	if le.isPrimary() {	// 检测当前的 lessor 是否为主 Lessor
		ls = le.findExpiredLeases(revokeLimit)	// 查询
	}
	le.mu.RUnlock()

	if len(ls) != 0 {
		select {
		case <-le.stopC:
			return
		case le.expiredC <- ls:	// 将过期的 Lease 实例集合
		default:
			// 如果当前 expiredC 通道阻塞，则放弃处理检测
		}
	}
}

// 检查点日程安排查找所有到期的预定租约检查点，并将其提交给检查点以将它们持久化到一致性日志。
func (le *lessor) checkpointScheduledLeases() {
	var cps []*pb.LeaseCheckpoint

	// 利率限制
	for i := 0; i < leaseCheckpointRate/2; i++ {
		le.mu.Lock()
		if le.isPrimary() {
			cps = le.findDueScheduledCheckpoints(maxLeaseCheckpointBatchSize)
		}
		le.mu.Unlock()

		if len(cps) != 0 {
			le.cp(context.Background(), &pb.LeaseCheckpointRequest{Checkpoints: cps})
		}
		if len(cps) < maxLeaseCheckpointBatchSize {
			return
		}
	}
}

func (le *lessor) clearScheduledLeasesCheckpoints() {
	le.leaseCheckpointHeap = make(LeaseQueue, 0)
}

func (le *lessor) clearLeaseExpiredNotifier() {
	le.leaseExpiredNotifier = newLeaseExpiredNotifier()
}

// 如果存在到期项，则expireExists返回true。
// 只有到期项存在时才会弹出。
// /“next”为真，表示它可能存在于下一次尝试中。
func (le *lessor) expireExists() (l *Lease, ok bool, next bool) {
	if le.leaseExpiredNotifier.Len() == 0 {
		return nil, false, false
	}

	item := le.leaseExpiredNotifier.Poll()
	l = le.leaseMap[item.id]
	if l == nil {
		// 租约已过期或已被撤销
		// 无需撤销（无到期）
		le.leaseExpiredNotifier.Unregister() // O（日志N）
		return nil, false, true
	}
	now := time.Now()
	if now.Before(item.time) /* item.time: expiration time */ {
		// 候选到期日已到，重新插入此项
		// 无需撤销（无到期）
		return l, false, false
	}

	// 重新检查重试间隔后撤销是否完成
	item.time = now.Add(le.expiredLeaseRetryInterval)
	le.leaseExpiredNotifier.RegisterOrUpdate(item)
	return l, true, false
}

// lessor.findExpiredLeases（）方法中会遍历 leaseMap 中全部的 Lease 实例，并调用
//Lease.expired（）方法检测其是否过期
func (le *lessor) findExpiredLeases(limit int) []*Lease {
	leases := make([]*Lease, 0, 16)	// 记录过期的 Lease 实例

	for {
		l, ok, next := le.expireExists()
		if !ok && !next {
			break
		}
		if !ok {
			continue
		}
		if next {
			continue
		}

		if l.expired() {	// 检测该 Lease 实例是否过期，其中通过当前时间与 Lease.expiry 进行比较
			leases = append(leases, l)

			// 达到过期限制
			if len(leases) == limit {
				break
			}
		}
	}

	return leases
}

func (le *lessor) scheduleCheckpointIfNeeded(lease *Lease) {
	if le.cp == nil {
		return
	}

	if lease.RemainingTTL() > int64(le.checkpointInterval.Seconds()) {
		if le.lg != nil {
			le.lg.Debug("Scheduling lease checkpoint",
				zap.Int64("leaseID", int64(lease.ID)),
				zap.Duration("intervalSeconds", le.checkpointInterval),
			)
		}
		heap.Push(&le.leaseCheckpointHeap, &LeaseWithTime{
			id:   lease.ID,
			time: time.Now().Add(le.checkpointInterval),
		})
	}
}

func (le *lessor) findDueScheduledCheckpoints(checkpointLimit int) []*pb.LeaseCheckpoint {
	if le.cp == nil {
		return nil
	}

	now := time.Now()
	cps := []*pb.LeaseCheckpoint{}
	for le.leaseCheckpointHeap.Len() > 0 && len(cps) < checkpointLimit {
		lt := le.leaseCheckpointHeap[0]
		if lt.time.After(now) /* lt.time: next checkpoint time */ {
			return cps
		}
		heap.Pop(&le.leaseCheckpointHeap)
		var l *Lease
		var ok bool
		if l, ok = le.leaseMap[lt.id]; !ok {
			continue
		}
		if !now.Before(l.expiry) {
			continue
		}
		remainingTTL := int64(math.Ceil(l.expiry.Sub(now).Seconds()))
		if remainingTTL >= l.ttl {
			continue
		}
		if le.lg != nil {
			le.lg.Debug("Checkpointing lease",
				zap.Int64("leaseID", int64(lt.id)),
				zap.Int64("remainingTTL", remainingTTL),
			)
		}
		cps = append(cps, &pb.LeaseCheckpoint{ID: int64(lt.id), Remaining_TTL: remainingTTL})
	}
	return cps
}

// 初始化操作
func (le *lessor) initAndRecover() {
	// 开启 v3 存储的读写事务，
	tx := le.b.BatchTx()

	tx.Lock()
	schema.UnsafeCreateLeaseBucket(tx)	// 创建 lease Bucket
	lpbs := schema.MustUnsafeGetAllLeases(tx)	// 查找 lease Bucket 中全部信息
	tx.Unlock()
	for _, lpb := range lpbs {
		ID := LeaseID(lpb.ID)
		if lpb.TTL < le.minLeaseTTL {	// 修正过期时间
			lpb.TTL = le.minLeaseTTL
		}
		le.leaseMap[ID] = &Lease{	// 创建 Lease 实例并记录到 leaseMap 中
			ID:  ID,
			ttl: lpb.TTL,
			// 恢复键值对时将填写项集
			// 将过期设置为永久，升级时刷新
			itemSet:      make(map[LeaseItem]struct{}),
			expiry:       forever,
			revokec:      make(chan struct{}),
			remainingTTL: lpb.RemainingTTL,
		}
	}
	le.leaseExpiredNotifier.Init()
	heap.Init(&le.leaseCheckpointHeap)

	le.b.ForceCommit()	// 提交事务
}

type Lease struct {
	// 该 Lease 实例的唯一标识。
	ID           LeaseID
	// 该 Lease 实例的存活时长。
	ttl          int64
	remainingTTL int64
	expiryMu sync.RWMutex
	// 该 Lease 实例过期的时间戳。
	expiry time.Time
	mu      sync.RWMutex
	// 该 map 中的 Key 是与当前 Lease 实例绑
	// 定的 Leaseltem 实例， Value 始终为空结构体
	itemSet map[LeaseItem]struct{}
	// 该 Lease 实例被撤销时会关 闭该通道，从而实现通
	// 知监听该通道的 goro u tine 的效果。
	revokec chan struct{}
}

func (l *Lease) expired() bool {
	return l.Remaining() <= 0
}

func (l *Lease) persistTo(b backend.Backend) {
	lpb := leasepb.Lease{ID: int64(l.ID), TTL: l.ttl, RemainingTTL: l.remainingTTL}
	tx := b.BatchTx()
	tx.Lock()
	defer tx.Unlock()
	schema.MustUnsafePutLease(tx, &lpb)
}

// TTL返回租约的TTL。
func (l *Lease) TTL() int64 {
	return l.ttl
}

// RemainingTTL返回租约的最后一个检查点剩余TTL。
// TODO（jpbetz）：不要公开此实用程序方法
func (l *Lease) RemainingTTL() int64 {
	if l.remainingTTL > 0 {
		return l.remainingTTL
	}
	return l.ttl
}

// refresh刷新租约到期时间。
func (l *Lease) refresh(extend time.Duration) {
	newExpiry := time.Now().Add(extend + time.Duration(l.RemainingTTL())*time.Second)
	l.expiryMu.Lock()
	defer l.expiryMu.Unlock()
	l.expiry = newExpiry
}

// forever将租赁期限设置为forever。
func (l *Lease) forever() {
	l.expiryMu.Lock()
	defer l.expiryMu.Unlock()
	l.expiry = forever
}

// Keys返回租约附带的所有密钥。
func (l *Lease) Keys() []string {
	l.mu.RLock()
	keys := make([]string, 0, len(l.itemSet))
	for k := range l.itemSet {
		keys = append(keys, k.Key)
	}
	l.mu.RUnlock()
	return keys
}

// 剩余返回租约的剩余时间。
func (l *Lease) Remaining() time.Duration {
	l.expiryMu.RLock()
	defer l.expiryMu.RUnlock()
	if l.expiry.IsZero() {
		return time.Duration(math.MaxInt64)
	}
	return time.Until(l.expiry)
}

type LeaseItem struct {
	Key string
}

func int64ToBytes(n int64) []byte {
	bytes := make([]byte, 8)
	binary.BigEndian.PutUint64(bytes, uint64(n))
	return bytes
}

// FakeLessor是出租方接口的伪实现。
// 仅用于测试。
type FakeLessor struct{}

func (fl *FakeLessor) SetRangeDeleter(dr RangeDeleter) {}

func (fl *FakeLessor) SetCheckpointer(cp Checkpointer) {}

func (fl *FakeLessor) Grant(id LeaseID, ttl int64) (*Lease, error) { return nil, nil }

func (fl *FakeLessor) Revoke(id LeaseID) error { return nil }

func (fl *FakeLessor) Checkpoint(id LeaseID, remainingTTL int64) error { return nil }

func (fl *FakeLessor) Attach(id LeaseID, items []LeaseItem) error { return nil }

func (fl *FakeLessor) GetLease(item LeaseItem) LeaseID            { return 0 }
func (fl *FakeLessor) Detach(id LeaseID, items []LeaseItem) error { return nil }

func (fl *FakeLessor) Promote(extend time.Duration) {}

func (fl *FakeLessor) Demote() {}

func (fl *FakeLessor) Renew(id LeaseID) (int64, error) { return 10, nil }

func (fl *FakeLessor) Lookup(id LeaseID) *Lease { return nil }

func (fl *FakeLessor) Leases() []*Lease { return nil }

func (fl *FakeLessor) ExpiredLeasesC() <-chan []*Lease { return nil }

func (fl *FakeLessor) Recover(b backend.Backend, rd RangeDeleter) {}

func (fl *FakeLessor) Stop() {}

type FakeTxnDelete struct {
	backend.BatchTx
}

func (ftd *FakeTxnDelete) DeleteRange(key, end []byte) (n, rev int64) { return 0, 0 }
func (ftd *FakeTxnDelete) End()                                       { ftd.Unlock() }
